/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data.trans.worker;

import cn.ac.iie.common.util.misc.MiscTools;
import cn.ac.iie.ulss.data.trans.commons.RuntimeEnv;
import cn.ac.iie.ulss.data.trans.util.Meta;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.log4j.Logger;

/**
 *
 * @author Li Mingyang
 */
public class AllConsumer extends Thread {
    
    static Logger logger = Logger.getLogger(AllConsumer.class.getName());
    
    @Override
    public void run() {
        Map<String, Meta> transMetaMap = (Map<String, Meta>) RuntimeEnv.getParam(RuntimeEnv.TRANSMETAMAP);
        Set<String> keys = transMetaMap.keySet();
        for (String key : keys) {
            Meta meta = transMetaMap.get(key);
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(meta.getGroupName());
            consumer.setNamesrvAddr(meta.getSourceParms());
            consumer.setInstanceName(meta.getGroupName() + MiscTools.getIpAddress() + "-" + MiscTools.getHostName());
            consumer.setConsumeConcurrentlyMaxSpan(5000);
            consumer.setPullThresholdForQueue(8);
            consumer.setPullBatchSize(8);
            consumer.setConsumeThreadMin(6);
            consumer.setConsumeThreadMax(9);
            consumer.setConsumeMessageBatchMaxSize(4);
            consumer.setClientCallbackExecutorThreads(24);
            consumer.setPullInterval(0);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            try {
                consumer.subscribe(meta.getSourceTopic(), "*");
            } catch (MQClientException ex) {
                logger.error(ex.getMessage(), ex);
            }
            Protocol protocol = null;
            try {
                protocol = Protocol.parse(new File("docs.json"));
            } catch (IOException ex) {
                logger.error(ex.getMessage(), ex);
            }
            final Schema docsSchema = protocol.getType("docs");
            final DatumReader<GenericRecord> docsReader = new GenericDatumReader(docsSchema);
            final String dataName = meta.getDataName();
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                        ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        byte[] binaryData = msg.getBody();
                        ByteArrayInputStream docsbis = new ByteArrayInputStream(binaryData);
                        BinaryDecoder docsbd = new DecoderFactory().binaryDecoder(docsbis, null);
                        GenericRecord docsRecord = new GenericData.Record(docsSchema);
                        try {
                            docsReader.read(docsRecord, docsbd);
                        } catch (IOException ex) {
                            logger.error(ex.getMessage(), ex);
                        }
                        GenericArray docSet = (GenericArray) docsRecord.get("doc_set");
                        while (!RuntimeEnv.TOPIC_TO_DATAPOOL.get(dataName).offer(binaryData)) {
                            logger.info("the LinkedBlockingQueue for " + dataName + " is full.");
                            try {
                                Thread.sleep(500);
                            } catch (InterruptedException ex) {
                                logger.error(ex.getMessage(), ex);
                            }
                        }
                        logger.info("get " + docSet.size() + " " + dataName);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            try {
                consumer.start();
            } catch (MQClientException ex) {
                logger.error(ex.getMessage(), ex);
            }
            RuntimeEnv.consumers.add(consumer);
        }
    }
}
